/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore.fieldmanager;

import java.lang.reflect.Array;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.Map;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.datanucleus.ClassLoaderResolver;
import org.datanucleus.ObjectManager;
import org.datanucleus.api.ApiAdapter;
import org.datanucleus.exceptions.NucleusDataStoreException;
import org.datanucleus.metadata.AbstractClassMetaData;
import org.datanucleus.metadata.AbstractMemberMetaData;
import org.datanucleus.store.ObjectProvider;
import cn.edu.thu.laud.objectstore.CassandraConsistency;
import cn.edu.thu.laud.utils.serde.ConverterContext;
import cn.edu.thu.laud.objectstore.index.BatchMutation;
import cn.edu.thu.laud.objectstore.index.IndexContainer;
import cn.edu.thu.laud.objectstore.index.IndexHandler;
import cn.edu.thu.laud.objectstore.index.IndexMeta;
import cn.edu.thu.laud.objectstore.index.KVColumn;
import cn.edu.thu.laud.utils.MetaDataUtils;
import org.datanucleus.store.fieldmanager.AbstractFieldManager;

/**
 * 
 * @author zhuoan
 *
 */
public class StoreFieldManager extends AbstractFieldManager {
	private AbstractClassMetaData metaData;
	private ObjectProvider objectProvider;
	private String cfName;
	private KVColumn kvColumn;
	private BatchMutation mutator;
	ConverterContext converterContext;
	Cassandra.Client client;
	ObjectManager om;
	IndexMeta idxMeta;

	public StoreFieldManager(ObjectManager om, ObjectProvider objcp,
			ConverterContext converterContext, Cassandra.Client client,
			AbstractClassMetaData acmd, KVColumn kvColumn, String cfName,
			BatchMutation mutator, IndexMeta idxMeta) {
		
		this.metaData = acmd;
		this.objectProvider = objcp;
		this.cfName = cfName;
		this.kvColumn = kvColumn;
		this.mutator = mutator;
		this.om = om;
		this.client = client;
		this.converterContext = converterContext;
		this.idxMeta = idxMeta;
	}

	public void storeBooleanField(int fieldNumber, boolean value) {

		try {
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			mutator.addInsertion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cfName }), new Column(colName)
							.setValue(converterContext.getBytes(value))
							.setTimestamp(System.currentTimeMillis()));
			
		
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	public void storeByteField(int fieldNumber, byte value) {

		try {
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			mutator.addInsertion(
					kvColumn.getRowKey(),
					Arrays.asList(new String[] { cfName }),
					new Column(colName).setValue(new byte[] { value })
							.setTimestamp(System.currentTimeMillis()));
			AbstractMemberMetaData memberData = metaData.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);
			String indexName = MetaDataUtils.getIndexName(memberData);
			if(indexName != null){
				IndexContainer container = new IndexContainer(idxMeta.getItem(), indexName);
				IndexHandler.updateCompositeIndex(client, mutator, kvColumn, ByteBufferUtil.string(colName), value, container, idxMeta, CassandraConsistency.get());
			}
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	public void storeCharField(int fieldNumber, char value) {

		try {			
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			mutator.addInsertion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cfName }), new Column(colName)
							.setValue(converterContext.getBytes(value))
							.setTimestamp(System.currentTimeMillis()));		
			
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	public void storeDoubleField(int fieldNumber, double value) {
		try {
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			mutator.addInsertion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cfName }), new Column(colName)
							.setValue(converterContext.getBytes(value))
							.setTimestamp(System.currentTimeMillis()));
			AbstractMemberMetaData memberData = metaData.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);
			String indexName = MetaDataUtils.getIndexName(memberData);
			if(indexName != null){
				IndexContainer container = new IndexContainer(idxMeta.getItem(), indexName);
				IndexHandler.updateCompositeIndex(client, mutator, kvColumn, ByteBufferUtil.string(colName), value, container, idxMeta, CassandraConsistency.get());
			}
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	public void storeFloatField(int fieldNumber, float value) {
		try {
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			mutator.addInsertion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cfName }), new Column(colName)
							.setValue(converterContext.getBytes(value))
							.setTimestamp(System.currentTimeMillis()));
			AbstractMemberMetaData memberData = metaData.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);
			String indexName = MetaDataUtils.getIndexName(memberData);
			if(indexName != null){
				IndexContainer container = new IndexContainer(idxMeta.getItem(), indexName);
				IndexHandler.updateCompositeIndex(client, mutator, kvColumn, ByteBufferUtil.string(colName), value, container, idxMeta, CassandraConsistency.get());
			}
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	public void storeIntField(int fieldNumber, int value) {
		try {
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			mutator.addInsertion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cfName }), new Column(colName)
							.setValue(converterContext.getBytes(value))
							.setTimestamp(System.currentTimeMillis()));
			AbstractMemberMetaData memberData = metaData.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);
			String indexName = MetaDataUtils.getIndexName(memberData);
			if(indexName != null){
				IndexContainer container = new IndexContainer(idxMeta.getItem(), indexName);
				IndexHandler.updateCompositeIndex(client, mutator, kvColumn, ByteBufferUtil.string(colName), value, container, idxMeta, CassandraConsistency.get());
			}
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	public void storeLongField(int fieldNumber, long value) {
		try {
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			mutator.addInsertion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cfName }), new Column(colName)
							.setValue(converterContext.getBytes(value))
							.setTimestamp(System.currentTimeMillis()));
			AbstractMemberMetaData memberData = metaData.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);
			String indexName = MetaDataUtils.getIndexName(memberData);
			if(indexName != null){
				IndexContainer container = new IndexContainer(idxMeta.getItem(), indexName);
				IndexHandler.updateCompositeIndex(client, mutator, kvColumn, ByteBufferUtil.string(colName), value, container, idxMeta, CassandraConsistency.get());
			}
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	public void storeShortField(int fieldNumber, short value) {
		try {
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			mutator.addInsertion(kvColumn.getRowKey(),
					Arrays.asList(new String[] { cfName }), new Column(colName)
							.setValue(converterContext.getBytes(value))
							.setTimestamp(System.currentTimeMillis()));
			AbstractMemberMetaData memberData = metaData.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);
			String indexName = MetaDataUtils.getIndexName(memberData);
			if(indexName != null){
				IndexContainer container = new IndexContainer(idxMeta.getItem(), indexName);
				IndexHandler.updateCompositeIndex(client, mutator, kvColumn, ByteBufferUtil.string(colName), value, container, idxMeta, CassandraConsistency.get());
			}
			
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	public void storeStringField(int fieldNumber, String value) {

		try {
			ByteBuffer colName = MetaDataUtils.getColumnName(metaData, fieldNumber);
			if (value == null) {
				SlicePredicate pred = new SlicePredicate();
				pred.setColumn_names(Arrays
						.asList(new ByteBuffer[] { colName }));
				mutator.addDeletion(
						kvColumn.getRowKey(),
						Arrays.asList(new String[] { cfName }),
						new Deletion().setPredicate(pred).setTimestamp(
								System.currentTimeMillis()));
				
			} else {

				mutator.addInsertion(kvColumn.getRowKey(),
						Arrays.asList(new String[] { cfName }), new Column(colName)
								.setValue(converterContext.getBytes(value))
								.setTimestamp(System.currentTimeMillis()));
			}
			
			AbstractMemberMetaData memberData = metaData.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);
			String indexName = MetaDataUtils.getIndexName(memberData);
			if(indexName != null){
				IndexContainer container = new IndexContainer(idxMeta.getItem(), indexName);
				IndexHandler.updateCompositeIndex(client, mutator, kvColumn, ByteBufferUtil.string(colName), value, container, idxMeta, CassandraConsistency.get());
			}
			
		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

	@Override
	public void storeObjectField(int fieldNumber, Object value) {
		try {

			ClassLoaderResolver clr = om.getClassLoaderResolver();
			AbstractMemberMetaData fieldMetaData = metaData
					.getMetaDataForManagedMemberAtAbsolutePosition(fieldNumber);

			int relationType = fieldMetaData.getRelationType(clr);
			if (fieldMetaData.isEmbedded()
					&& MetaDataUtils.isRelationSingleValued(relationType)) {

				throw new NucleusDataStoreException(
						"Embedded objects are unsupported.  Mark the object as persistent and use a serializable class instead");

			}
			// check if this is a relationship
			ByteBuffer columnName = MetaDataUtils.getColumnName(metaData,
					fieldNumber);
			if (MetaDataUtils.isRelationSingleValued(relationType)) {
				// Persistable object - persist the related object and store the
				// identity in the cell
				if (value == null) {
					SlicePredicate pred = new SlicePredicate();
					pred.setColumn_names(Arrays
							.asList(new ByteBuffer[] { columnName }));
					this.mutator.addDeletion(
							kvColumn.getRowKey(),
							Arrays.asList(new String[] { cfName }),
							new Deletion().setPredicate(pred).setTimestamp(
									System.currentTimeMillis()));
					return;
				}

				Object persisted = om.getExecutionContext()
						.persistObjectInternal(value, objectProvider,
								fieldNumber, 0);

				Object objectPk = om.getApiAdapter()
						.getIdForObject(persisted);

				mutator.addInsertion(kvColumn.getRowKey(), Arrays.asList(new String[]{cfName}),
						new Column(columnName).setValue(
								converterContext.getRowKeyForId(objectPk)).setTimestamp(System.currentTimeMillis()));

				return;
			}

			if (MetaDataUtils.isRelationMultiValued(relationType)) {
				// Collection/Map/Array

				if (fieldMetaData.hasCollection()) {

					WriteCollection collectionWriter = new WriteCollection(
							client, converterContext, cfName, kvColumn.getRowKey(),
							columnName);

					if (value == null) {
						collectionWriter.removeAllColumns(mutator);
						return;
					}

					Object persisted = null;
					Object objectPk = null;

					for (Object element : (Collection<?>) value) {
						// persist the object
						persisted = om.getExecutionContext().persistObjectInternal(element,
								objectProvider, fieldNumber, 0);

						objectPk = om.getApiAdapter().getIdForObject(
								persisted);

						collectionWriter.writeRelationship(mutator, objectPk);
					}

					// TODO remove this when SCO is working
					collectionWriter.removeRemaining(mutator);

					objectProvider.wrapSCOField(fieldNumber, value, true, true,
							true);

					return;

				} else if (fieldMetaData.hasMap()) {

					WriteMap mapWriter = new WriteMap(client, converterContext,
							cfName, kvColumn.getRowKey(), columnName);

					if (value == null) {
						mapWriter.removeAllColumns(mutator);
						return;
					}

					ApiAdapter adapter = om.getApiAdapter();

					Map<?, ?> map = ((Map<?, ?>) value);

					// serialized values to store per item
					Object serializedKey = null;
					Object serializedValue = null;

					// value set by the user in the map
					Object mapValue = null;

					// pointer to what we persisted
					Object persisted = null;

					// get each element and persist it.
					for (Object mapKey : map.keySet()) {

						mapValue = map.get(mapKey);

						// handle the case if our key is a persistent class
						// itself
						if (adapter.isPersistable(mapKey)) {

							persisted = om.getExecutionContext().persistObjectInternal(mapKey,
									objectProvider, fieldNumber, 0);

							serializedKey = om.getApiAdapter()
									.getIdForObject(persisted);
						} else {
							serializedKey = mapKey;
						}

						// persist the value if it can be persisted
						if (adapter.isPersistable(mapValue)) {

							persisted = om.getExecutionContext().persistObjectInternal(mapValue,
									objectProvider, fieldNumber, 0);

							serializedValue = om.getApiAdapter()
									.getIdForObject(persisted);

						} else {
							serializedValue = mapValue;
						}

						mapWriter.writeRelationship(mutator, serializedKey,
								serializedValue);

						// TODO remove this when SCO is working
						mapWriter.removeRemaining(mutator);

					}

					objectProvider.wrapSCOField(fieldNumber, value, true, true,
							true);

					return;

				} else if (fieldMetaData.hasArray()) {

					Object persisted = null;
					Object objectPk = null;

					WriteMap mapWriter = new WriteMap(client, converterContext,
							cfName, kvColumn.getRowKey(), columnName);

					if (value == null) {
						mapWriter.removeAllColumns(mutator);
						return;
					}

					for (int i = 0; i < Array.getLength(value); i++) {

						// persist the object
						persisted = om.getExecutionContext().persistObjectInternal(
								Array.get(value, i), objectProvider,
								fieldNumber, 0);

						objectPk = om.getApiAdapter().getIdForObject(
								persisted);

						mapWriter.writeRelationship(mutator, i, objectPk);
					}

					// TODO remove this when SCO is working
					mapWriter.removeRemaining(mutator);
				}

				objectProvider.wrapSCOField(fieldNumber, value, true, true,
						true);

				return;
			}

			if (value == null) {
				SlicePredicate pred = new SlicePredicate();
				pred.setColumn_names(Arrays
						.asList(new ByteBuffer[] { columnName }));
				this.mutator.addDeletion(
						kvColumn.getRowKey(),
						Arrays.asList(new String[] { cfName }),
						new Deletion().setPredicate(pred).setTimestamp(
								System.currentTimeMillis()));				
			} else {
	
				ByteBuffer data = converterContext.getBytes(value);
				mutator.addInsertion(kvColumn.getRowKey(), Arrays.asList(new String[]{cfName}),
						new Column(columnName).setValue(data).setTimestamp(System.currentTimeMillis()));
			}
			
			String indexName = MetaDataUtils.getIndexName(fieldMetaData);
			if(indexName != null){
				Object idxValue = null;
				if(value != null)
					idxValue = converterContext.getIndexValue(value);
				if(value != null && idxValue == null)
					throw new NucleusDataStoreException("Unsupported index value type "+ value.getClass());
				IndexContainer container = new IndexContainer(idxMeta.getItem(), indexName);
				IndexHandler.updateCompositeIndex(
						client,
						mutator,
						kvColumn,
						ByteBufferUtil.string(columnName),
						value == null ? null : idxValue, container, idxMeta, CassandraConsistency.get());
			}

		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
	}

}